
#include "TCPConnection.h"
#include "Socket.h"
#include "sock.h"
#include "eventloop/Eventloop.h"
#include "eventloop/IOchannel.h"
#include "log/Logger.h"
#include "utility/assertion.h"
#include "utility/cast.h"
#include "wrap/error.h"
#include "wrap/unistd.h"


TCPConnection::TCPConnection(
    Eventloop &loop, const std::string &name, std::unique_ptr<Socket> &&socket,
    const Sockaddr &local, const Sockaddr &peer):
    std::enable_shared_from_this<TCPConnection>{},
    _inputBuffer{},
    _outputBuffer{},
    _name{name},
    _closeCallback{default_connection_func},
    _connectionCallback{default_connection_func},
    _drainCallback{},
    _highWaterCallback{},
    _messageCallback{default_message_func},
    _local{local},
    _peer{peer},
    _loop{loop},
    _socket{std::move(socket)},
    _channel{},
    _highWaterMark{1 << (16 + 8 + 2)},
    _state{_State::_CONNECTING} {
    _channel = std::make_unique<IOchannel>(_loop, _socket->fd(), name);
    _channel->set_close_callback([this](){ _close_callback(); });
    _channel->set_error_callback([this](){ _error_callback(); });
    _channel->set_read_callback(
        [this](Utility::timepoint t){ _read_callback(t); });
    _channel->set_write_callback([this](){ _write_callback(); });
    LOG_DEBUG << "TCPConnection [" << name << "] at " << this
              << " fd=" << _channel->fd();
    _socket->keep_alive(true);
}
                  
                  
TCPConnection::~TCPConnection() {
    LOG_DEBUG << "TCPConnection [" << _name << "] at " << this 
              << " fd=" << _channel->fd() << " state=" << _state_to_string();
    assert(_state == _State::_DISCONNECTED);
}


void TCPConnection::set_connection_callback(const ConnectionCallback &callback) {
    assert(callback);
    _connectionCallback = callback;
}

void TCPConnection::set_close_callback(const CloseCallback &callback) {
    assert(callback);
    _closeCallback = callback;
}

void TCPConnection::set_message_callback(const MessageCallback &callback) {
    assert(callback);
    _messageCallback = callback;
}

void TCPConnection::set_drain_callback(const DrainCallback &callback) {
    if (!callback) {
        _drainCallback.reset();
        return;
    }
    _drainCallback = callback;
}

void TCPConnection::set_high_water_mark_callback(const HighWaterMarkCallback &callback)
{
    if (!callback) {
        _highWaterCallback.reset();
        return;
    }
    _highWaterCallback = callback;
}

void TCPConnection::established() {
    _loop.ASSERT_IN_THIS_THREAD();
    assert(_state == _State::_CONNECTING);
    _state = _State::_CONNECTED;
    _channel->tie(shared_from_this());
    _channel->enable_reading();
    _connectionCallback(shared_from_this());
}

void TCPConnection::destroy() {
    _loop.ASSERT_IN_THIS_THREAD();
    if (_state == _State::_CONNECTED) {
        _state = _State::_DISCONNECTED;
        _channel->disable();
        _connectionCallback(shared_from_this());
    }
    _channel->remove();
}

void TCPConnection::force_close() {
    if (_state != _State::_CONNECTED && _state != _State::_DISCONNECTING) {
        return;
    }
    _state = _State::_DISCONNECTING;
    _loop.queue_in_loop([sharedThis = shared_from_this()]() {
        sharedThis->_loop.ASSERT_IN_THIS_THREAD();
        if (sharedThis->_state != _State::_CONNECTED &&
            sharedThis->_state != _State::_DISCONNECTING) {
            return;
        }
        sharedThis->_close_callback();
    });
}

const char *TCPConnection::_state_to_string() const {
    switch (_state) {
    case _State::_DISCONNECTED:
        return "_DISCONNECTED";
        break;
    case _State::_CONNECTING:
        return "_CONNECTING";
        break;
    case _State::_CONNECTED:
        return "_CONNECTED";
        break;
    case _State::_DISCONNECTING:
        return "_DISCONNCETING";
        break;
    default:
        return "UNKNOWN STATE";
    }
}

void TCPConnection::_read_callback(Utility::timepoint t) {
    _loop.ASSERT_IN_THIS_THREAD();
    int savedErrno = 0;
    auto n = _inputBuffer.read_fd(_channel->fd(), savedErrno);
    if (n > 0) {
        _messageCallback(shared_from_this(), _inputBuffer, t);
        return;
    }
    else if (n == 0) {
        _close_callback();
        return;
    }
 
    errno = savedErrno;
    LOG_SYSERROR << "TCPConnection::read_fd<0";
    _error_callback();
}

void TCPConnection::_write_callback() {
    _loop.ASSERT_IN_THIS_THREAD();
    if (!_channel->is_writing()) {
        LOG_VERBOSE << "Connection fd=" << _channel->fd() << " is down";
        return;
    }
    const auto n = Wrap::write_(
        _channel->fd(), _outputBuffer.peek(), _outputBuffer.readable_bytes());
    if (n <= 0) {
        LOG_SYSERROR << "TCPConnection write " << n << "<=0";
        return;
    }
    _after_write(static_cast<std::size_t>(n));
}

void TCPConnection::_error_callback() {
    const auto err = Sock::socket_error(_channel->fd());
    LOG_ERROR << "TCPConnection [" << _name << "]-ERROR=" << err << ' '
              << Wrap::strerror_r_(err);
}

void TCPConnection::_close_callback() {
    _loop.ASSERT_IN_THIS_THREAD();
    LOG_VERBOSE << "fd=" << _channel->fd() << " state=" << _state_to_string();
    assert(_state == _State::_CONNECTED || _state == _State::_DISCONNECTING);
    _state = _State::_DISCONNECTED;
    _channel->disable();
    const auto guard = shared_from_this();
    _connectionCallback(guard);
    _closeCallback(guard);
}

void TCPConnection::send(const std::string &message){
    if (!_can_send_in_other_loop(message)) {
        return;
    }
    _loop.run_in_loop([this, msg = message](){ _send_in_loop(msg); });
}

void TCPConnection::send(std::string &&message) {
    if (!_can_send_in_other_loop(message)) {
        return;
    }
    _loop.run_in_loop(
        [this, msg = std::move(message)](){ _send_in_loop(msg); });
}

void TCPConnection::shutdown() {
    if (_state != _State::_CONNECTED) {
        return;
    }
    _state = _State::_DISCONNECTING;
    _loop.run_in_loop([this](){ _shut_down_in_loop(); });
}

void TCPConnection::_after_write(std::size_t written)
{
    _outputBuffer.retrieve(written);
    if (_outputBuffer.readable_bytes() != 0) {
        return;
    }
    _channel->disable_writing();
    if (_drainCallback.has_value()) {
        _loop.queue_in_loop([sharedThis = shared_from_this()]() {
            sharedThis->_drainCallback.value()(sharedThis);
        });
    }
    if (_state != _State::_DISCONNECTING) {
        return;
    }
    _shut_down_in_loop();
}

bool TCPConnection::_can_send_in_other_loop(const std::string_view &view)
{
    if (_state != _State::_CONNECTED) {
        return false;
    }
    if (_loop.is_in_this_thread()) {
        _send_in_loop(view);
        return false;
    }
    return true;
}

TCPConnection::_SendResult TCPConnection::_send(const std::string_view &view) {
    _SendResult result{view.size(), 0, false};
    if (_channel->is_writing() || _outputBuffer.readable_bytes() != 0) {
        return result;
    }
    result.written = Wrap::write_(_channel->fd(), view.data(), view.size());
    if (result.written >= 0) {
        result.remain = Cast::SUB(view.size(), result.written);
        if (result.remain == 0 && _drainCallback.has_value()) {
            _loop.queue_in_loop([sharedThis = shared_from_this()](){
                sharedThis->_drainCallback.value()(sharedThis);
            });
        }
    }
    else {
        result.written = 0;
        if (errno != EWOULDBLOCK) {
            LOG_SYSERROR << "TCPConnection errno " << errno;
            if (errno == EPIPE || errno == ECONNRESET) {
                result.faultError = true;
            }
        }
    }
    return result;
}

void TCPConnection::_send_in_loop(const std::string_view &view){
    _loop.ASSERT_IN_THIS_THREAD();
    if (_state == _State::_DISCONNECTED) {
        LOG_WARN << "disconnected, give up writing";
        return;
    }
    const auto sendResult = _send(view);
    assert(sendResult.remain <= view.size());
    if (sendResult.faultError || sendResult.remain <= 0) {
        return;
    }
    _water_mark_func(sendResult.remain);
    _outputBuffer.append(view.data() + sendResult.written, sendResult.remain);
    if (_channel->is_writing()) {
        return;
    }
    _channel->enable_writing();

}

void TCPConnection::_shut_down_in_loop() const {
    _loop.ASSERT_IN_THIS_THREAD();
    if (_channel->is_writing()) {
        return;
    }
    _socket->shutdown_write();
}

void TCPConnection::_water_mark_func(std::size_t remain) {
    const auto oldSize = _outputBuffer.readable_bytes();
    const auto water = oldSize + remain;
    if (water >= _highWaterMark && oldSize < _highWaterMark && 
        _highWaterCallback.has_value()) {
        _loop.queue_in_loop([sharedThis = shared_from_this(), water](){
            sharedThis->_highWaterCallback.value()(sharedThis, water);
        });
    }
}
